package mmq.nio;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.channels.Channels;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//http://today.java.net/cs/user/print/a/350
public class Server {

	private final Selector selector;
	private final ServerSocketChannel serverChannel;
	private final ExecutorService threadPool;
	private boolean running;
	
	private Server() throws IOException {
		System.out.println("Server.Server()");
		selector = Selector.open();
		serverChannel = ServerSocketChannel.open();
		serverChannel.configureBlocking(false);
		threadPool = Executors.newFixedThreadPool(10);
	}
	
	private int init() throws IOException {
		System.out.println("Server.init()");
		serverChannel.socket().bind(null);
		running = true;
		return serverChannel.socket().getLocalPort();
	}
	
	private void start() throws IOException {
		System.out.println("Server.start()");
		threadPool.submit(new Runnable() {
			public void run() {
				System.out.println("Acceptor");
				try {
					serverChannel.register(selector, SelectionKey.OP_ACCEPT);
					while(running) {
						selector.select();
						Set<SelectionKey> keys = selector.selectedKeys();
						for(SelectionKey key : keys) {
							keys.remove(key);
							if(key.isValid() && key.isAcceptable()) {
								assert serverChannel == key.channel();
								final SocketChannel channel = serverChannel.accept();
								if(channel != null) {
									System.out.println(String.format("Accepted: %s", channel));
									threadPool.submit(new Runnable() {
										public void run() {
											try {
												handle(channel);
											} catch(IOException e) {
												throw new RuntimeException(e);
											}
										}});
								}
							}
						}
					}
				} catch(IOException e) {
					throw new RuntimeException(e);
				}
			}});
	}
	
	private void handle(SocketChannel channel) throws IOException {
		System.out.println("Server.handle()");
		channel.register(selector, SelectionKey.OP_READ);
		while(running) {
			if(selector.select() > 0) {
				Set<SelectionKey> keys = selector.selectedKeys();
				for(SelectionKey key : keys) {
					System.out.println(String.format("Read key: %s", key));
					keys.remove(key);
					if(key.isValid() && key.isReadable()) {
						InputStream is = Channels.newInputStream(channel);
						int i = is.read();
						while(i > -1) {
							System.out.println(Integer.toHexString(i));
						}
					}
				}
			}
		}
	}
	
	public static void main(String[] args) throws Exception {
		Server server = new Server();
		try {
			int port = server.init();
			server.start();
			server.call(port, new byte[] {13, 87, 23, 124});
		} finally {
			//server.shutdown();
		}
	}
	
	private void shutdown() throws IOException {
		try {
			running = false;
			serverChannel.close();
		} finally {
			threadPool.shutdown();
		}
	}

	private void call(final int port, final byte[] content) {
		threadPool.submit(new Runnable() {
			public void run() {
				Socket socket = new Socket();
				try {
					socket.connect(new InetSocketAddress("localhost", port));
					try {
						OutputStream os = socket.getOutputStream();	
						os.write(content);
						os.flush();
					} finally {
						socket.close();
					}
				} catch (IOException e) {
					throw new RuntimeException(e);
				}
			}});
	}
	
}
